package com.zz.study.sharding.task;

import com.zz.study.sharding.algorithm.compress.DPUtil;
import com.zz.study.sharding.component.StarRocksStreamWriter;
import com.zz.study.sharding.constants.Constants;
import com.zz.study.sharding.mappers.SlaveDtFusdataFlinkMapper;
import com.zz.study.sharding.util.ConcurrentUtil;
import com.zz.study.sharding.util.DateUtil;
import com.zz.study.sharding.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;

import static com.zz.study.sharding.constants.Constants.CompressHisTrackDataTableNamePrefix;
import static com.zz.study.sharding.constants.Constants.HisTrackDataTableNamePrefix;

/**
 * created by xjx on  2024-01-26 10:32
 */
@Slf4j
public class TrackQueryTask implements Callable<List<Map<String,Object>> > {


    private SlaveDtFusdataFlinkMapper slaveDtFusdataFlinkMapper;

    private List<String> tarIdList;

    private Date startTime;

    private Date endTime;

    private double dsMax;

    private StarRocksStreamWriter starRocksStreamWriter;


    public TrackQueryTask(SlaveDtFusdataFlinkMapper slaveDtFusdataFlinkMapper, List<String> tarIdList, Date startTime, Date endTime,double dsMax,StarRocksStreamWriter starRocksStreamWriter) {
        this.slaveDtFusdataFlinkMapper = slaveDtFusdataFlinkMapper;
        this.tarIdList = tarIdList;
        this.startTime = startTime;
        this.endTime = endTime;
        this.dsMax = dsMax;
        this.starRocksStreamWriter = starRocksStreamWriter;
    }

    @Override
    public List<Map<String,Object>>  call() throws Exception {
        List<Map<String,Object>> compressListTotal = new ArrayList<>();

        List<Map<String,Object>> dataList = new ArrayList<>();
        Map<String,List<Map<String,Object>>> map = new HashMap<>();
        try {
            String targetTable = DateUtil.getShardTableName(HisTrackDataTableNamePrefix,endTime);
            dataList =   slaveDtFusdataFlinkMapper.getTrackMapByTarIdListAndTime2(targetTable,tarIdList,startTime,endTime);
            log.error("根据tarId数组获取数据====="+dataList.size()+"targetTable=="+targetTable+"startTime="+DateUtil.formatDate(startTime)+"endTime==="+DateUtil.formatDate(endTime));
            for(Map<String,Object> bean:dataList){
                String tarId = bean.get("TarID").toString();
                if(!map.containsKey(tarId)){
                    List tempList = new ArrayList();
                    tempList.add(bean);
                    map.put(tarId,tempList);
                }else{
                    List tempList = map.get(tarId);
                    tempList.add(bean);
                    map.put(tarId,tempList);
                }
            }
            log.error("map的长度====="+map.size());
        }catch (Exception e){
            log.error("并发根据tarId获取轨迹错误======="+e.getMessage());
        }
       for(String key:map.keySet()){
           List<Map<String,Object>> tempList = map.get(key);
           List<Map<String,Object>> afterCompressList =  DPUtil.dpAlgorithmMap(tempList,dsMax);
           if(!CollectionUtils.isEmpty(afterCompressList)){
               compressListTotal.addAll(afterCompressList);
           }
       }
//        log.error("每个tarId对应的数据，进行压缩");
//        try{
//            String flushData = JsonUtil.serialize(compressListTotal);
//            if(StringUtils.isNotEmpty(flushData)){
//                starRocksStreamWriter.importBeanMapList2Doris(compressListTotal, targetCompressDataTableName, Constants.trackDataColumns);
//                log.error("写入数据到starRocks，compressListTotal的长度"+compressListTotal.size());
//                Thread.sleep(1000);
//            }
//        }catch (Exception e){
//            e.printStackTrace();
//            log.error(e.getMessage());
//        }
            return compressListTotal;
    }
}
